package com.pangu.mq.kafka.consumer;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.core.task.TaskExecutor;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.CompletableFuture;

import static java.lang.System.currentTimeMillis;

/**
 * @author Linbizhao
 * @description:
 * @since 2023/11/23 15:20
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class UpstreamMessageConsumer {

    private final TaskExecutor upstreamTaskExecutor;

    @KafkaListener(containerFactory = "", topics = "", groupId = "", concurrency = "")
    public void handlerLimiterMessage(List<ConsumerRecord<String, String>> records) {
        final long start = currentTimeMillis();
        CompletableFuture.allOf(records.stream().map(r -> CompletableFuture.runAsync(() -> this.handleMessage(r), upstreamTaskExecutor))
                        .toArray(CompletableFuture[]::new))
                .thenRun(() -> log.info("【消息总线】- UpstreamMessage消息批量处理完成，处理数量:[{}]，costTime:[{}]ms", records.size(), currentTimeMillis() - start));
    }


    public void handleMessage(ConsumerRecord<String, String> record) {
        val s = currentTimeMillis();
        val messageJsonStr = record.value();
        // TODO: 2024/2/29  
    }
}
